















#include <assert.h>
#include <sys/mman.h>
#include <stdio.h>
#include <stdlib.h>

#include "Errors.h"
#include "DataTypes.h"
#include "FileLoader.h"
#include "PageConversions.h"
#include "DiskArray.h"
#include "MetadataDB.h"
#include <iostream>
#include <vector>
#include <algorithm>
using namespace std;

class objectHolder{
        public:
                INT indicies;
                FLOAT value;

       friend  bool operator< (const objectHolder& A,const  objectHolder& B);
       friend  bool operator> (const objectHolder& A, const objectHolder& B);
};
bool operator< (const objectHolder& a,const objectHolder& b){
        return (a.value<b.value);
}
bool operator> (const objectHolder& a,const objectHolder& b){
        return (a.value>b.value);
}


FileLoaderImp::FileLoaderImp(const char* _fileName, EventProcessor& _fileWriter, DistributedCounter* _dCounter):
	chunksProduced(0), dCounter(_dCounter), tuplesHandled(0)
#ifdef  DEBUG_EVPROC
	,EventProcessorImp(true, "FileLoader")
#endif
{
	fileName = strdup(_fileName);
	finished=false;

	fileWriter.CopyFrom(_fileWriter);
	DiskArray& diskArray = DiskArray::GetDiskArray();
	pageSize = diskArray.getPageSize();

	// allocate memory for the buffers

					    	  
  l_orderkeyData=(IDENTIFIER*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_partkeyData=(IDENTIFIER*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_supkeyData=(IDENTIFIER*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_linenumberData=(INT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(INT)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_quantityData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_extendedpriceData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_discountData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_taxData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_returnflagData=(CHAR<1>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_linestatusData=(CHAR<1>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_shipdateData=(DATE*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_commitdateData=(DATE*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_receiptdateData=(DATE*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_shipinstructData=(CHAR<25>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<25>)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			    	  
  l_shipmodeData=(CHAR<10>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<10>)), 
		  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  			      
  // allocating memory for  column l_comment of type VARCHAR
  l_commentData=(long int *) mmap(NULL, bytesToPageAlligned(pageSize, 2097152*sizeof(long int)), 
		PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
  
  l_commentContent=(char*) mmap(NULL, bytesToPageAlligned(pageSize, 2097152*256), 
		       PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);
     	
	
  l_indiciesData = (INT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(INT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

 l_sortedkeyData = (IDENTIFIER*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedpartkeyData=(IDENTIFIER*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedsupkeyData=(IDENTIFIER*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedlinenumberData=(INT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(INT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedquantityData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedextendedpriceData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sorteddiscountData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedtaxData=(FLOAT*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedreturnflagData=(CHAR<1>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedlinestatusData=(CHAR<1>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedshipdateData=(DATE*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedcommitdateData=(DATE*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedreceiptdateData=(DATE*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedshipinstructData=(CHAR<25>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<25>)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedshipmodeData=(CHAR<10>*) mmap(NULL, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<10>)),
                  PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

                                                              

 l_sortedcommentData=(long int *) mmap(NULL, bytesToPageAlligned(pageSize, 2097152*sizeof(long int)),
                PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);

  l_sortedcommentContent=(char*) mmap(NULL, bytesToPageAlligned(pageSize, 2097152*256),
                       PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, 0, 0);




	// register messages
	RegisterMessageProcessor(ChunkWritten::type, &ChunkWrittenProc, 1);

	// open the file and get ready to read
	stream = fopen(fileName, "r");
	if ( stream==NULL){
		printf( "File %s could not be opened in FileLoader:", fileName);
		perror(NULL);
	}

	// start the whole thing by sending the first chunk
	FillAndSendChunk();
}

FileLoaderImp::~FileLoaderImp(void){
					    
  munmap(l_orderkeyData, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)));
  			    
  munmap(l_partkeyData, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)));
  			    
  munmap(l_supkeyData, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)));
  			    
  munmap(l_linenumberData, bytesToPageAlligned( pageSize, 2097152*sizeof(INT)));
  			    
  munmap(l_quantityData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));
  			    
  munmap(l_extendedpriceData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));
  			    
  munmap(l_discountData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));
  			    
  munmap(l_taxData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));
  			    
  munmap(l_returnflagData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)));
  			    
  munmap(l_linestatusData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)));
  			    
  munmap(l_shipdateData, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)));
  			    
  munmap(l_commitdateData, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)));
  			    
  munmap(l_receiptdateData, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)));
  			    
  munmap(l_shipinstructData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<25>)));
  			    
  munmap(l_shipmodeData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<10>)));
  			    
  // freeing memory for   column l_comment of type VARCHAR
  munmap(l_commentData, bytesToPageAlligned(pageSize, 2097152*sizeof(long int)));
  munmap(l_commentContent,  bytesToPageAlligned(pageSize, 2097152*256));
     	
	
  munmap(l_sortedkeyData, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)));

  munmap(l_sortedpartkeyData, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)));

  munmap(l_sortedsupkeyData, bytesToPageAlligned( pageSize, 2097152*sizeof(IDENTIFIER)));

  munmap(l_sortedlinenumberData, bytesToPageAlligned( pageSize, 2097152*sizeof(INT)));

  munmap(l_sortedquantityData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));

  munmap(l_sortedextendedpriceData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));

  munmap(l_sorteddiscountData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));

  munmap(l_sortedtaxData, bytesToPageAlligned( pageSize, 2097152*sizeof(FLOAT)));

  munmap(l_sortedreturnflagData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)));

  munmap(l_sortedlinestatusData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<1>)));

  munmap(l_sortedshipdateData, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)));

  munmap(l_sortedcommitdateData, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)));

  munmap(l_sortedreceiptdateData, bytesToPageAlligned( pageSize, 2097152*sizeof(DATE)));

  munmap(l_sortedshipinstructData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<25>)));

  munmap(l_sortedshipmodeData, bytesToPageAlligned( pageSize, 2097152*sizeof(CHAR<10>)));
  munmap(l_sortedcommentData, bytesToPageAlligned(pageSize, 2097152*sizeof(long int)));
  munmap(l_sortedcommentContent,  bytesToPageAlligned(pageSize, 2097152*256));


	free(fileName);
	fclose(stream);
}




bool sortByValue(const objectHolder&A, const objectHolder& B){
        return (A.value < B.value);
}

void FileLoaderImp::Done(void){
	cout <<  "Producer " << fileName << "DONE\n";
	cout << "Have handled :" << tuplesHandled << " Tuples" << endl;
	
		
	// get the relationID
		
	// update the freeChunkID of the relation
			

	if (dCounter->Decrement(1) == 0){
		// this is the last producer, kill the writer
		printf("I am the last Generator. I'm killing the Writer\n");
		
		// this is the last producer, kill the adder
		DieMessage_Factory(fileWriter);

		sleep(1);
		KillEvProc(fileWriter);
	}
}



//int compare(const void* a, const void *b){ return (*(int*)a-*(int*)b);}

void FileLoaderImp::FillAndSendChunk(void){
	int noTuples=0;
	char* curr;
	char* next;

	size_t lineSize; // the size of the line
	char* buffer = NULL; // will be resized by getline

					      			      			      			      			      			      			      			      			      			      			      			      			      			      			      			    
  // aux variable for column l_comment
  char* l_commentCurr=l_commentContent;
     	
	
	while(true){
		// get next line of text
		// if eof break
		size_t lineRead = getline(&buffer, &lineSize, stream);
		if ( lineRead == -1){
			finished=true; // we read the whole file
			break; // we reached the end of file
		}

		next = buffer;
		
		//cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;

									// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_orderkeyData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_partkeyData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_supkeyData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_linenumberData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_quantityData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_extendedpriceData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_discountData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_taxData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_returnflagData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_linestatusData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_shipdateData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_commitdateData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_receiptdateData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_shipinstructData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
      FromString(*(l_shipmodeData+noTuples), curr);
  					// tokenizing
			curr=next;
			while( (next-buffer)< lineSize && *next != '|' && *next!='\n') { next++; }
			*next = 0;
			next++;
			// cout << *(l_orderkeyData+noTuples) << " " <<  l_orderkeyData+noTuples << " " << noTuples << endl;


			// column filling code
			    
     *(l_commentData+noTuples)=(long int)(l_commentCurr-l_commentContent);
     strcpy(l_commentCurr, curr);
     l_commentCurr+=strlen(l_commentCurr)+1;
     		
		
		noTuples++;
		tuplesHandled++;

		if (noTuples>=2097152)
			break;
	}

vector<objectHolder> listOfData;
objectHolder objectIdent;

for(int i = 0; i < noTuples; i++){

        objectIdent.value = *(l_taxData+i);
        objectIdent.indicies = i;
	cout << " taxData" << *(l_taxData+i) << endl;
        listOfData.push_back(objectIdent);
}

        sort(listOfData.begin(), listOfData.end(), sortByValue);

int j = 0;
while(j < noTuples){

        *(l_indiciesData+j) = *(l_linenumberData+listOfData[j].indicies);
        *(l_sortedkeyData+j) = *(l_orderkeyData+listOfData[j].indicies);

        *(l_sortedpartkeyData) = *(l_partkeyData+listOfData[j].indicies); // column l_partkey array                                    
        *(l_sortedsupkeyData) = *(l_supkeyData+listOfData[j].indicies); // column l_supkey array                              
        *(l_sortedlinenumberData) = *(l_linenumberData+listOfData[j].indicies); // column l_linenumber array                                     
        *(l_sortedquantityData) = *(l_quantityData+listOfData[j].indicies); // column l_quantity array                               
        *(l_sortedextendedpriceData) = *(l_extendedpriceData+listOfData[j].indicies); // column l_extendedprice array                                     
        *(l_sorteddiscountData) = *(l_discountData+listOfData[j].indicies); // column l_discount array                               
        *(l_sortedtaxData) = *(l_taxData+listOfData[j].indicies); // column l_tax array                                 
        *(l_sortedreturnflagData) = *(l_returnflagData+listOfData[j].indicies);
        *(l_sortedlinestatusData) = *(l_linestatusData+listOfData[j].indicies);
        *(l_sortedshipdateData) = *(l_shipdateData+listOfData[j].indicies);
        *(l_sortedcommitdateData) = *(l_commitdateData+listOfData[j].indicies);
        *(l_sortedreceiptdateData) = * (l_receiptdateData+listOfData[j].indicies); // column l_receiptdate array                                  
        *(l_sortedshipinstructData) =   *(l_shipinstructData+listOfData[j].indicies);
        *(l_sortedshipmodeData) =  *(l_shipmodeData+listOfData[j].indicies);

   	*(l_sortedcommentData) = *(l_commentData+listOfData[j].indicies);
        *(l_sortedcommentContent) = *(l_commentContent+listOfData[j].indicies);

	cout << "value: " << listOfData[j].value << " index: " << listOfData[j].indicies <<  "sorted " << *(l_sortedkeyData+j) << endl;
j++;
}
	l_linenumberData = l_sortedlinenumberData;
	l_orderkeyData = l_sortedkeyData;
	l_partkeyData = l_sortedpartkeyData;
	l_supkeyData = l_sortedsupkeyData;
	l_extendedpriceData = l_sortedextendedpriceData;
	l_discountData = l_sorteddiscountData;
	l_taxData = l_sortedtaxData;
	l_returnflagData = l_sortedreturnflagData;
	l_linestatusData = l_sortedlinestatusData;
	l_shipdateData = l_sortedshipdateData;
	l_commitdateData = l_sortedcommitdateData;
	l_receiptdateData =l_sortedreceiptdateData;
	l_commitdateData = l_sortedcommitdateData;
	l_receiptdateData = l_sortedreceiptdateData;
	l_shipinstructData = l_sortedshipinstructData;
	l_shipmodeData = l_sortedshipmodeData;	
	l_commentData = l_sortedcommentData;
	l_commentContent = l_sortedcommentContent;
	// deallocate the buffer (not needed anymore)
	free(buffer);

	// we might be finished and not produce any tuples
	// if so we do not want to sent a request to write an empty chunk to the disk
	if (finished && noTuples == 0){
		Done();
		return;
	}

//temp disk



//merge


//send


	// send the request
	ColWriteDescriptionContainer cols;

					      
  // record column l_orderkey
  {
	  ColWriteDescription
      colFixed(l_orderkeyData, noTuples*sizeof(IDENTIFIER));
    cols.Append(colFixed);
  }
  			      
  // record column l_partkey
  {
	  ColWriteDescription
      colFixed(l_partkeyData, noTuples*sizeof(IDENTIFIER));
    cols.Append(colFixed);
  }
  			      
  // record column l_supkey
  {
	  ColWriteDescription
      colFixed(l_supkeyData, noTuples*sizeof(IDENTIFIER));
    cols.Append(colFixed);
  }
  			      
  // record column l_linenumber
  {
	  ColWriteDescription
      colFixed(l_linenumberData, noTuples*sizeof(INT));
    cols.Append(colFixed);
  }
  			      
  // record column l_quantity
  {
	  ColWriteDescription
      colFixed(l_quantityData, noTuples*sizeof(FLOAT));
    cols.Append(colFixed);
  }
  			      
  // record column l_extendedprice
  {
	  ColWriteDescription
      colFixed(l_extendedpriceData, noTuples*sizeof(FLOAT));
    cols.Append(colFixed);
  }
  			      
  // record column l_discount
  {
	  ColWriteDescription
      colFixed(l_discountData, noTuples*sizeof(FLOAT));
    cols.Append(colFixed);
  }
  			      
  // record column l_tax
  {
	  ColWriteDescription
      colFixed(l_taxData, noTuples*sizeof(FLOAT));
    cols.Append(colFixed);
  }
  			      
  // record column l_returnflag
  {
	  ColWriteDescription
      colFixed(l_returnflagData, noTuples*sizeof(CHAR<1>));
    cols.Append(colFixed);
  }
  			      
  // record column l_linestatus
  {
	  ColWriteDescription
      colFixed(l_linestatusData, noTuples*sizeof(CHAR<1>));
    cols.Append(colFixed);
  }
  			      
  // record column l_shipdate
  {
	  ColWriteDescription
      colFixed(l_shipdateData, noTuples*sizeof(DATE));
    cols.Append(colFixed);
  }
  			      
  // record column l_commitdate
  {
	  ColWriteDescription
      colFixed(l_commitdateData, noTuples*sizeof(DATE));
    cols.Append(colFixed);
  }
  			      
  // record column l_receiptdate
  {
	  ColWriteDescription
      colFixed(l_receiptdateData, noTuples*sizeof(DATE));
    cols.Append(colFixed);
  }
  			      
  // record column l_shipinstruct
  {
	  ColWriteDescription
      colFixed(l_shipinstructData, noTuples*sizeof(CHAR<25>));
    cols.Append(colFixed);
  }
  			      
  // record column l_shipmode
  {
	  ColWriteDescription
      colFixed(l_shipmodeData, noTuples*sizeof(CHAR<10>));
    cols.Append(colFixed);
  }
  			    
  // record column l_comment	      
  {
     ColWriteDescription
       colVar(l_commentData, noTuples*sizeof(long int), l_commentContent, 
	   l_commentCurr-l_commentContent, true /** we have offsets not pointers */);
     cols.Append(colVar);
  }
     	
	
	// if we do not clone the interface, the method swaps the content out
	EventProcessor myInterfClone;
	myInterfClone.CopyFrom(myInterface);
	ChunkWrite_Factory( fileWriter, chunksProduced, noTuples, myInterfClone, cols);

	cout << "Loader of  " << fileName <<" : chunk " << chunksProduced << " out" << endl;

	chunksProduced++;
}

MESSAGE_HANDLER_DEFINITION_BEGIN(FileLoaderImp, ChunkWrittenProc, ChunkWritten){

	cout << "Loader of  " << evProc.fileName << " : chunk " << evProc.chunksProduced << " BACK " << endl;

	if (!evProc.finished){
		evProc.FillAndSendChunk();
	} else {
		evProc.Done();
	}
}MESSAGE_HANDLER_DEFINITION_END
